package com.gt.ln

import com.gt.SCUtil
import org.apache.commons.lang3.StringUtils
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD

//先按照 clickCount 排序，再按照 bookCount，第三按照payCount ，最后按照id排序
case class product(id: String, clickCount :Int, bookCount :Int, payCount :Int) extends Ordered[product] with Serializable{
  override def compare(that: product): Int = {
    if (this.clickCount > that.clickCount){
      1
    }else if(this.clickCount < that.clickCount){
      -1
    }else{
      if(this.bookCount > that.bookCount){
        1
      }else if(this.bookCount < that.bookCount){
        -1
      }else{
        if (this.payCount > that.payCount) {
          1
        } else if (this.payCount < that.payCount) {
          -1
        } else {
          id.toInt - that.id.toInt
        }
      }
    }
  }
}

object Spark_hot_product_02 {

  def main(args: Array[String]): Unit = {
    val sc: SparkContext = SCUtil.createLocalSc()

    //1. 获取数据
    val rdd: RDD[String] = sc.textFile("data/user_visit_action.txt")

    //val clickCategoryId = arr(6) //点击类 品类id
    //val bookCategoryId = arr(8) //下单类 品类id a-b-c
    //val payCategoryId = arr(10) //支付类 品类id a-b-c
    //2. 转换数据格式
    val rdd2: RDD[(String, (Int, Int, Int))] = rdd.map(line => line.split("_"))
      .map(arr => {
        val clickCategoryId = arr(6) //点击类 品类id
        val bookCategoryId = arr(8) //下单类 品类id a-b-c
        val payCategoryId = arr(10) //支付类 品类id a-b-c

        if (StringUtils.isNotEmpty(clickCategoryId) && clickCategoryId.toInt != -1 && !"null".equals(clickCategoryId)) {
          (clickCategoryId, (1, 0, 0))
        } else if (StringUtils.isNotEmpty(bookCategoryId) && !"null".equals(bookCategoryId)) {
          (bookCategoryId, (0, 1, 0))
        } else if (StringUtils.isNotEmpty(payCategoryId) && !"null".equals(payCategoryId)) {
          (payCategoryId, (0, 0, 1))
        } else {
          ("", (0, 0, 0))
        }
      }).filter(data => StringUtils.isNotEmpty(data._1) && !data._1.equals("null"))

    val rdd3: RDD[(String, (Int, Int, Int))] = rdd2.map(data => {
        data._1.split(",").map(id => List((id, data._2)))
      }).flatMap(data => data)
      .flatMap(data => data)

    val rdd4: RDD[product] = rdd3.reduceByKey((a, b) => {
      (a._1 + b._1, a._2 + b._2, a._3 + b._3)
    }).map(data => {
      product(data._1, data._2._1, data._2._2, data._2._3)
    }).sortBy(p => p)




    rdd4.collect().reverse.foreach(println)
    sc.stop()
  }

}
